package com.chengqs.rainbow.c2c.service.impl;

import com.chengqs.rainbow.common.core.domain.BasePage;
import com.chengqs.rainbow.common.im.constant.Constants;
import com.chengqs.rainbow.common.im.domain.MessageJsonPush;
import com.chengqs.rainbow.common.im.domain.MessageSendJsonRequest;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.collections4.Get;
import org.apache.commons.collections4.Put;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.stream.Collectors;

/**
 * <B>主类名称：</B>MessageSendServiceImpl<BR>
 * <B>概要说明：</B>enter<BR>
 *
 * @author: chengqs
 * @create: 2025-09-16 23:03:27
 **/
@Slf4j
@RequiredArgsConstructor
@Service
public class MessageSendServiceImpl extends AbstractMessageSendService {

    private static final Integer MESSAGE_FETCH_THRESHOLD = 10;

    private static final int FACTOR = 128;

    @Override
    public void save(List<MessageSendJsonRequest> requests, Long chatId) {
        List<Put> puts = new ArrayList<>(requests.size());

        Map<Long, List<MessageSendJsonRequest>> groupByToId = requests.stream().collect(Collectors.groupingBy(MessageSendJsonRequest::getToId));
        for (Map.Entry<Long, List<MessageSendJsonRequest>> entry : groupByToId.entrySet()) {
            if (entry.getValue().size() > MESSAGE_FETCH_THRESHOLD) {
                sendMessageFetch(chatId, entry.getKey());
                for (MessageSendJsonRequest request : entry.getValue()) {
                    sendMessageSendResponse(request);
                }
            } else {
                for (MessageSendJsonRequest request : entry.getValue()) {
                    sendMessageSendResponse(request);
                    sendMessagePush(request);
                }
            }
        }
    }

    @Override
    public BasePage<MessageJsonPush> fetch(Long chatId, Long startMessageId, Long stopMessageId, int size) {

        return new BasePage<>();
    }

    private String getRowKey(Long toId, Long messageId) {
        Long tempMessageId = messageId;
        if (Objects.isNull(tempMessageId)) {
            tempMessageId = 0L;
        }
        // 逆序消息id，做排序用，最新的消息-先查找
        return getRowKey(toId)
                + StringUtils.leftPad(String.valueOf(Long.MAX_VALUE - tempMessageId), 19, '0');
    }

    private String getRowKey(Long toId) {
        int hash = (int) (toId % FACTOR);
        // 3位会话id的hash+群id+逆序消息id
        // hash，是为了在hbase中分区存储，相当于mysql的分表
        // 群id，rowKey做群消息查询
        return StringUtils.leftPad(String.valueOf(hash), 3, '0') + "|"
                + StringUtils.leftPad(String.valueOf(toId), 19, '0') + "|";
    }
}
